package com.wuba.rxjavademo.lyz;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.observables.GroupedObservable;
import rx.schedulers.Schedulers;

/**
 * Created by liyaxing on 2016/4/14.
 */
public class Demo04 {

    public static void main(String[] ss) throws InterruptedException {

        Demo04 demo04 = new Demo04() ;
//        Observable<String> flatMapObservable = demo04.flatMapObservable() ;
//        flatMapObservable.subscribe(o -> System.out.println("flatMapObservable o = " + o)) ;
//        Observable<String> flatMapIterableObservble = demo04.flatMapIterableObservble() ;
//        flatMapIterableObservble.subscribe(o -> System.out.println("flatMapIterableObservble o = " + o)) ;
//        Observable<GroupedObservable<String, Integer>> groupByObservable = demo04.groupByObservable() ;
//        groupByObservable.subscribe(stringIntegerGroupedObservable -> {
//            String key = stringIntegerGroupedObservable.getKey() ;
//            if("A".equals(key)){
//                stringIntegerGroupedObservable.subscribe(integer -> {
//                    System.out.println("key = " + key + " ; integer = " + integer) ;
//                }) ;
//            }
//        }) ;
//        Observable<Integer> castObservable = demo04.castObservable().cast(Integer.class) ;
//        castObservable.subscribe(integer -> System.out.println("integer = " + integer)) ;
//        Observable<Integer> scanObservable = demo04.scanObservable() ;
//        scanObservable.subscribe(integer -> System.out.println("integer = " + integer)) ;
//        Observable<Integer> throttleWithTimeoutObservable = demo04.throttleWithTimeoutObservable() ;
//        throttleWithTimeoutObservable.subscribe(integer -> System.out.println("integer = " + integer)) ;

//        Observable<Integer> observable = demo04.debounceObserver() ;
//        observable.subscribe(integer -> {
//            System.out.println("debounce : " + integer) ;
//        });
//        Thread.sleep(10000);
//        Observable<Observable<String>> observable = demo04.groupJoinObservable() ;
//        observable.subscribe(obj -> {
//            System.out.println("result : " + obj) ;
//            obj.subscribe(s ->  System.out.println("string : " + s)) ;
//        });
//        Thread.sleep(100000);

        Observable<Observable<Integer>> observable = demo04.zipObservable() ;
        observable.subscribe(obj -> {
//            System.out.println("result : " + obj) ;
            obj.subscribe(integer -> System.out.println("result : " + integer)) ;
        });




        Thread.sleep(100000);


    }

    private Observable<Observable<Integer>> zipObservable(){

        List<Observable<Integer>> list = new ArrayList<>() ;
        for (int i = 0; i < 5; i++) {
            list.add(combineLastObservable1(i)) ;
        }

        return Observable.zip(list,args -> {

            int total = 0 ;
            for (int i = 0; i < args.length; i++) {
                System.out.println("args["+i+"] = " + args[i]);
                total += (Integer)args[i] ;
            }

            int _total = total ;

            return Observable.<Integer>create(subscriber -> subscriber.onNext(_total)) ;
        }) ;
    }

    private Observable<Observable<Integer>> zipWithObservable(){
        return combineLastObservable1(1).zipWith(combineLastObservable1(2),(num1,num2) -> {
            System.out.println("num1 = " + num1);
            System.out.println("num2 = " + num2);
            return Observable.<Integer>create(subscriber -> subscriber.onNext(num1 * num2)) ;
        }) ;
    }

    private Observable<Integer> switchObservable(){
        return Observable.switchOnNext(Observable.create(subscriber -> {
            for (int i = 0; i < 5; i++) {
                subscriber.onNext(combineLastObservable1(i));
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        })) ;
    }

    private Observable<Integer> startWithObservable(){
        return Observable.just(1,2,3,4).startWith(-1,-2) ;
    }

    private Observable<Integer> mergeDelayErrorObservable(){
        return Observable.<Integer>mergeDelayError(Observable.create(subscriber -> {
            for (int i = 0; i < 5; i++) {
                if(i == 3){
                    subscriber.onError(new Throwable("error")) ;
                }
                subscriber.onNext(i);
            }
        }),Observable.create(subscriber -> {
            for (int i = 0; i < 5; i++) {
                if(i == 3){
                    subscriber.onError(new Throwable("error")) ;
                }
                subscriber.onNext(i);
            }
        })) ;
    }

    private Observable<Integer> mergeObservable(){
        return Observable.merge(Observable.just(1, 23, 4, 56, 6).subscribeOn(Schedulers.newThread()), Observable.just(2,3,4,5,4)) ;
    }

    private Observable<Observable<String>> groupJoinObservable(){
        return Observable.just("Left").groupJoin(combineLastObservable1(1),
                s -> Observable.timer(3000,TimeUnit.SECONDS),
                i -> Observable.timer(2000,TimeUnit.SECONDS),
                (left,observable) -> observable.map(s -> left + s)) ;
    }

    private Observable<String> joinObservable(){
        return Observable.just("Source").join(combineLastObservable1(1),
                s -> Observable.timer(3000,TimeUnit.SECONDS),
                s -> Observable.timer(2000,TimeUnit.SECONDS),
                (left,right) -> left + right) ;
    }

    private Observable<Integer> combineLastObservable(){
        return Observable.combineLatest(combineLastObservable1(1),combineLastObservable1(2),combineLastObservable1(3),(x,y,z)->{
            System.out.println("x : " + x);
            System.out.println("y : " + y);
            System.out.println("z : " + z);
            return x * y ;
        }) ;
    }

    private Observable<Integer> combineListObservable(){
        List<Observable<Integer>> list = new ArrayList<>() ;
        for (int i = 0; i < 5; i++) {
            list.add(combineLastObservable1(i)) ;
        }
        return Observable.combineLatest(list,args -> {
            int num = 0 ;
            for (int i = 0; i < args.length; i++) {
                System.out.println("args["+i+"] : " + args[i]);
                num += (Integer)args[i] ;
            }
            return num ;
        }) ;
    }

    private Observable<Integer> combineLastObservable1(int index){
        return Observable.<Integer>create(subscriber -> {
            for (int i = 1; i < 9; i++) {
                subscriber.onNext(i * index);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
//                    e.printStackTrace();
                }
            }
            subscriber.onCompleted();
        }).subscribeOn(Schedulers.newThread()) ;
    }

    private Observable<Integer> sampleObservable(){
        return Observable.<Integer>create(subscriber -> {
            for (int i = 0; i < 1000; i++) {
                subscriber.onNext(i);
                try {
                    Thread.sleep(220);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            subscriber.onCompleted();
        }).sample(1000,TimeUnit.MILLISECONDS) ;
    }

    private Observable<Integer> takeObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).take(3) ;
    }

    private Observable<Integer> skipObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).skip(3) ;
    }

    private Observable<Integer> lastObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).last(integer -> integer > 10) ;
    }

    private Observable<Integer> firstObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).first(integer -> integer > 10) ;
    }

    private Observable<Integer> filterObervable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).filter(integer -> integer > 10) ;
    }

    private Observable<Integer> elementAtObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).elementAt(2) ;
    }

    private Observable<Integer> distinctUtilChangedObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).distinctUntilChanged() ;
    }

    private Observable<Integer> distinctObservable(){
        return Observable.just(1,22,22,34,55,55,65,55,3,1).distinct() ;
    }

    private Observable<Integer> debounceObserver(){
        return Observable.just(1,2,3,4,5,6,7,8,9).debounce(integer -> Observable.<String>create(subscriber -> {
            if(integer % 2 == 0 && !subscriber.isUnsubscribed()){
                subscriber.onNext(integer + "::A");
//                subscriber.onCompleted();
            }
        })) ;
    }

    private Observable<Integer> throttleWithTimeoutObservable(){
        return Observable.<Integer>create(subscriber -> {
            for (int i = 0; i < 10; i++) {
                if(!subscriber.isUnsubscribed()){
                    subscriber.onNext(i);
                }
                long sleepTime = 100 ;
                if(i % 3 == 0){
                    sleepTime = 300 ;
                }

                try {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            subscriber.onCompleted();
        }).throttleWithTimeout(200,TimeUnit.MILLISECONDS) ;
    }

    private Observable<Observable<Long>> windowTimeObservable(){
        return Observable.interval(1, TimeUnit.SECONDS).window(3,TimeUnit.SECONDS) ;
    }

    private Observable<Observable<Integer>> windowObservable(){
        return Observable.just(1,2,3,4,5,6,7,8,9).window(3) ;
    }

    private Observable<Integer> scanObservable(){
        return Observable.just(1,2,3,4,5,6,7,8,9).scan((x,y) -> y) ;
    }

    private Observable<Object> castObservable(){
        return Observable.just(1,2,3,4,5) ;
    }

    private Observable<Integer> mapObservable(){
        return Observable.just(1,2,3,4,5).map(integer -> integer * 5 ) ;
    }

    private Observable<GroupedObservable<String, Integer>> groupByObservable(){
        return Observable.just(1,2,3,4,5).groupBy(integer -> {
//            System.out.println("groupByObservable : integer = " + integer) ;
            if(integer % 2 == 0){
//                System.out.println("groupByObservable : return A") ;
                return "A" ;
            }else {
//                System.out.println("groupByObservable : return B") ;
                return "B" ;
            }
        }) ;
    }

    private Observable<String> flatMapObservable(){
        return Observable.just(1,2,3,4,5).flatMap(integer -> Observable.just("" + integer)) ;
    }

    private Observable<String> flatMapIterableObservble(){
        return Observable.just(1,2,3,4,5,6).flatMapIterable(integer -> {
            ArrayList<String> list = new ArrayList<>() ;
            for (int i = 0; i < integer; i++) {
                list.add(integer + "") ;
            }
            return list ;
        }) ;
    }

}
